package com.pw.study.flink.chapter8;

import com.pw.study.flink.entities.Ads;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;

public class ExerciseAds {


    public static void main(String[] args) {
        exercise();
    }

    private static void exercise() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);


        //输入数据
        SingleOutputStreamOperator<Ads> stream = env.readTextFile("data/file/AdClickLog.csv").map(ads -> {
            String[] line = ads.split(",");
            return new Ads(Long.valueOf(line[0]), Long.valueOf(line[1]), line[2], line[3], Long.valueOf(line[4])*1000);
        });
        //
        WatermarkStrategy<Ads> strategy = WatermarkStrategy.<Ads>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((SerializableTimestampAssigner<Ads>) (ads, ts) -> ads.getTimestamp());

        stream.assignTimestampsAndWatermarks(strategy).keyBy(ads -> ads.getUserId() + "_" + ads.getAdsId())
                .process(new KeyedProcessFunction<String, Ads, String>() {
                    //昨天
                    private ValueState<String> yesterdayState;
                    //黑名单
                    private ValueState<Boolean> blackState;
                    //
                    private ReducingState<Long> clickState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        clickState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Long>("clickState", Long::sum, Long.class));
                        blackState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("blackState",Boolean.class));
                        yesterdayState = getRuntimeContext().getState(new ValueStateDescriptor<>("yesterdayState", String.class));
                    }

                    @Override
                    public void processElement(Ads value, Context ctx, Collector<String> out) throws Exception {
                        // 一旦跨天, 状态应该重置
                        // 如何判断数据跨天?  判断年月日是否变化
                        String yesterday = yesterdayState.value();
                        String today = new SimpleDateFormat("yyyy-MM-dd").format(value.getTimestamp());
                        if (!today.equals(yesterday)) {  // 表示数据进入到第二天
                            yesterdayState.update(today);  // 把今天的日期存入到状态, 明天就可以读到今天是昨天
                            clickState.clear();
                            blackState.clear();

                        }
                        // 如果已经进入了黑名单, 则不需要再进行统计
                        if (blackState.value() == null) {
                            clickState.add(1L);
                        }
                        Long count = clickState.get();
                        String msg = "用户: " + value.getUserId() + "对广告:" + value.getAdsId() + "的点击量是: " + count;

                        if (count > 99) {
                            // 第一次超过99加入黑名单, 以后就不用再加
                            if (blackState.value() == null) {
                                msg += " 超过阈值 99, 加入黑名单";
                                out.collect(msg);
                                blackState.update(true);
                            }
                        } else {
                            out.collect(msg);

                        }
                    }
                }).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
